package org.niit.service

import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.AdClickData
import org.niit.common.TService
import org.niit.dao.ZhangRealTimeDao

import java.io.{FileWriter, PrintWriter,File}
import java.text.SimpleDateFormat
import java.time.{DayOfWeek, LocalDateTime}
import java.time.format.DateTimeFormatter
import java.util.{Date,Locale}
import scala.util.matching.Regex
import scala.collection.mutable.ListBuffer

class ZhangTimeService extends TService[DStream[AdClickData]]{

  override def dataAnalysis(data: DStream[AdClickData]): Unit = {

    realTime1(data)
    realTime2(data)
    realTime3(data)
    realTime4(data)
    realTime5(data)
    realTime6(data)
  }

  //实时监测新用户的涌入情况，计算每5秒新用户占比变化。   x: 时间  y: 新用户占比
//  def realTimeNewUserRatio(data: DStream[AdClickData]): Unit = {
//    val mappedDS = data.map { line =>
//      val (total, newUser) = if (line.new_user == "1") (1, 1) else (1, 0)
//      ("global", (total, newUser))  // 固定key，全局累计
//    }
//    val cumulativeDS = mappedDS.reduceByKeyAndWindow(
//      (a: (Int, Int), b: (Int, Int)) => (a._1 + b._1, a._2 + b._2),
//      Seconds(24 * 3600),
//      Seconds(10)
//    )
//    val resultDS = cumulativeDS.map { case ("global", (total, newUser)) =>
//      //根据时间进行格式化，只保留  时：分：秒
//      val sdf = new SimpleDateFormat("HH-mm-ss")
//      val timeKey = sdf.format(new Date(Locale.US.toString))
//      val rate = (newUser.toDouble / total * 100).formatted("%.2f").toDouble
//      (timeKey,total,newUser,rate)
//    }
//    resultDS.foreachRDD(rdd => {
//      val list = new ListBuffer[String]()
//      val sortedData = rdd.collect()
//      sortedData.foreach {
//        case (timeKey,total,newUser,rate) => {
//          list.append(s"""{"xtime": "$timeKey", "yval1": "$total", "yval2": "$newUser", "yval3": "$rate"}""")
//          // 写入数据库（累计占比）
//          val dao = new ZhangRealTimeDao
//          dao.insertNewUserRatio(timeKey,total.toString,newUser.toString,rate)
//        }
//      }
//      // 写入JSON文件（略）
//      val out = new PrintWriter(new FileWriter("output/ZhangTime1.json"))
//      out.println("[\n"+list.mkString(",\n")+ ",\n]")
//      out.flush()
//      out.close()
//    })
//  }

  //一、实时监测每5秒新用户占比变化
  def realTime1(data: DStream[AdClickData]):Unit ={
    // 1. 数据映射与窗口聚合（，按用户去重）
    val userData = data.map(line => {
      val newuser = line.new_user == "1"
      ((line.time.toLong / 10000 * 10000, newuser, line.device),1)
      }).reduceByKeyAndWindow((_+_),Seconds(60 * 60), Seconds(5))

    // 2. 计算新用户占比并整理yval结构
    val resultData = userData.map {
      case ((time, newuser, device),count) => ((time,device),count)
      }.reduceByKeyAndWindow((_+_),Seconds(60 * 60 ),Seconds(5))

    val rs = resultData.map {
      case ((time, device), count) => (device, count)
    }.reduceByKeyAndWindow((_ + _), Seconds(60 * 60), Seconds(5))

    // 3. 写入文件（按新格式加载数据，确保累加不覆盖）
    rs.foreachRDD(rdd => {
      val list = new ListBuffer[String]()
      // 添加新数据（按要求格式：{"xtime": "...", "yval": "...."）
      val newData = rdd.sortByKey(true).collect()  // 按时间排序新数据
      newData.foreach {
        case (device, count) =>
          list.append(s"""{"xtime": "$device", "yval": $count}""")
          val zhangRealTimeDao = new ZhangRealTimeDao
          zhangRealTimeDao.insertTime1(device,count)
      }

      // 写入完整JSON数组（包含历史+新数据）
      val out = new PrintWriter(new FileWriter("output/zhangTime1.json"))
      out.println("[\n" + list.mkString(",\n") + "\n]")
      out.flush()
      out.close()
    })
  }

  //二、用户在不同设备上完成三步行为的实时完成率
  def realTime2(data: DStream[AdClickData]):Unit ={
    // 输出文件路径
    val outputFile = "output/zhangTime2.json"

    // 1. 数据映射：提取用户行为、设备及时间，按（设备）分组
    val userBehavior = data.map(line => {
        val userId = line.user_id  // 用户ID，用于关联行为
        val device = line.device   // 设备类型，用于分组
        // 标记用户是否访问过首页、列表页、产品页
        val hasHome = line.home_page == "1"
        val hasList = line.listing_page == "1"
        val hasProduct = line.product_page == "1"
        ( device, (userId, hasHome, hasList, hasProduct))
      }).groupByKey()

    // 2. 统计各设备的三步行为指标
    val windowedData = userBehavior.map{ case (device, userBehaviors) =>
      // 去重用户，确保每个用户只统计一次
      val uniqueUsers = userBehaviors.groupBy(_._1).mapValues(_.head).values  // 按userId去重，取第一条行为
      // 统计访问首页的用户数
      val homeTotal = uniqueUsers.count(_._2)  // _2是hasHome
      // 统计完成三步行为的用户数（同时满足首页、列表页、产品页）
      val threeStepTotal = uniqueUsers.count(u => u._2 && u._3 && u._4)  // _2:home, _3:list, _4:product
      (device, homeTotal, threeStepTotal)
    }

    // 3. 计算三步完成率
    val resultData = windowedData.map{
      case (device, homeTotal, threeStepTotal) =>
      // 计算完成率（避免除零，首页用户数为0时完成率为0）
      val completionRate = (threeStepTotal.toDouble / homeTotal * 100).formatted("%.2f")
      (device, homeTotal.toString, threeStepTotal.toString, completionRate)
    }

    // 4. 写入文件（累加存储，按时间+设备展示）
    resultData.foreachRDD(rdd => {
      val list = new ListBuffer[String]()
      // 添加新数据（按时间排序，字段包含时间、设备、完成率等）
      val newData = rdd.sortBy(_._1).collect()  // 按时间升序排列
      newData.foreach {
        case (device, homeTotal, threeStepTotal, rate) =>
        list.append(s"""{"device": "$device", "homeTotal": "$homeTotal", "threeStepTotal": "$threeStepTotal", "completionRate": "$rate"}""")
        val zhangRealTimeDao = new ZhangRealTimeDao
        zhangRealTimeDao.insertTime2(device, homeTotal, threeStepTotal, rate)
      }

      // 写入完整JSON数组（累加历史数据，不覆盖）
      val out = new PrintWriter(new FileWriter(outputFile))
      out.println("[\n" + list.mkString(",\n") + "\n]")
      out.flush()
      out.close()
    })

  }

  //三、不同用户在不同设备完成支付的实时支付率
  def realTime3(data: DStream[AdClickData]):Unit ={
    // 输出文件路径
    val outputFile = "output/zhangTime3.json"

    // 1. 数据映射：提取用户行为、分组维度及时间
    val userBehavior = data.map(line => {
        val userId = line.user_id  // 用户ID，用于去重
        val userType = if (line.new_user == "1") "new" else "old"  // 新/老用户标记
        val os = line.operative_system  // 操作系统
        val timeStamp = line.time.toLong
        val windowKey = timeStamp / 15000 * 15000  // 15秒窗口键
        // 标记是否访问产品页和支付确认页
        val hasProduct = line.product_page == "1"
        val hasPaymentConfirm = line.payment_confirmation_page == "1"
        // 组合键：(用户类型, 操作系统)；值：(用户ID, 产品页访问标记, 支付确认页访问标记)
        ((userType, os), (userId, hasProduct, hasPaymentConfirm))
      })
      // 按组合键聚合，关联同一窗口、同一用户类型、同一操作系统的用户行为
      .groupByKey()

    // 2. 统计转化指标（去重用户）
    val windowedData = userBehavior.map { case ((userType, os), userBehaviors) =>
      // 去重用户，每个用户只统计一次（取其行为标记）
      val uniqueUsers = userBehaviors.groupBy(_._1).mapValues(_.head).values
      // 统计访问产品页的用户数（分母）
      val productTotal = uniqueUsers.count(_._2)  // _2是hasProduct
      // 统计访问产品页且完成支付确认的用户数（分子）
      val paymentConfirmTotal = uniqueUsers.count(u => u._2 && u._3)  // _2:product, _3:paymentConfirm
      (userType, os, productTotal, paymentConfirmTotal)
    }

    // 3. 计算转化效率
    val resultData = windowedData.map { case (userType, os, productTotal, paymentTotal) =>
      // 计算转化率（避免除零，产品页用户数为0时转化率为0）
      val conversionRate = if (productTotal > 0) (paymentTotal.toDouble / productTotal * 100).formatted("%.2f")  else 0.0
      (userType, os, productTotal.toString, paymentTotal.toString, conversionRate)
    }

    // 4. 写入文件（累加存储，按时间排序）
    resultData.foreachRDD(rdd => {
      val list = new ListBuffer[String]()

      // 添加新数据（按时间排序，保持键值格式）
      val newData = rdd.sortBy(_._1).collect()  // 按时间升序排列
      newData.foreach { case (userType, os, productTotal, paymentTotal, rate) =>
        list.append(s"""{"userType": "$userType", "os": "$os", "productTotal": "$productTotal", "paymentConfirmTotal": "$paymentTotal", "conversionRate": "$rate"}""")
        val zhangRealTimeDao = new ZhangRealTimeDao
        zhangRealTimeDao.insertTime3(userType, os, productTotal, paymentTotal, rate.toString)
      }

      // 写入完整JSON数组（累加历史数据，不覆盖）
      val out = new PrintWriter(new FileWriter(outputFile))
      out.println("[\n" + list.mkString(",\n") + "\n]")
      out.flush()
      out.close()
    })
  }

  //四、不同性别年龄用户访问列表页的实时次数
  def realTime4(data: DStream[AdClickData]):Unit ={
    // 输出文件路径
    val outputFile = "output/zhangTime4.json"

    // 1. 数据映射：处理年龄区间、提取关键字段
    val userBehavior = data.map(line => {
        val userId = line.user_id  // 用户ID，用于去重
        val sex = line.sex  // 性别
        val ageStr = line.age
        // 年龄区间划分（处理非数字年龄为“未知”）
        val ageGroup = try {
          val age = ageStr.toInt
          if (age >= 17 && age <= 22) "17-22岁"
          else if (age >= 23 && age <= 30) "23-30岁"
          else if (age >= 31) "31岁+"
          else "未知"
        } catch {
          case _: Exception => "未知"
        }
        val source = line.source  // 来源渠道
        // 列表页访问标记（1为访问，0为未访问）及总浏览页数
        val hasListing = if (line.listing_page == "1") 1 else 0
        val totalPages = try { line.total_pages_visited.toInt } catch { case _: Exception => 0 }
        // 组合键：(性别, 年龄区间)；值：(用户ID, 列表页访问标记, 总浏览页数, 来源渠道)
        ((sex, ageGroup), (userId, hasListing, totalPages, source))
      })
      // 按组合键聚合，关联同一窗口、性别、年龄区间的用户行为
      .groupByKey()

    // 2. 统计分组指标（去重用户）
    val windowedData = userBehavior.map { case ((sex, ageGroup), userBehaviors) =>
      // 按用户ID去重，聚合用户行为（取该用户的总列表页访问次数、总浏览页数、来源渠道）
      val userAgg = userBehaviors
        .groupBy(_._1)  // 按userId分组
        .map { case (uid, behaviors) =>
          val totalListingVisits = behaviors.map(_._2).sum  // 该用户的列表页总访问次数
          val totalPages = behaviors.map(_._3).sum  // 该用户的总浏览页数
          (uid, totalListingVisits, totalPages)
        }

      // 分组总指标
      val totalUsers = userAgg.size  // 该分组的去重用户数
      val listingUsers = userAgg.count(_._2 > 0)  // 访问过列表页的用户数
      val totalListingVisits = userAgg.map(_._2).sum  // 该分组的列表页总访问次数
      val avgListingVisits = if (listingUsers > 0) (totalListingVisits.toDouble / listingUsers).formatted("%.2f") else "0.00"  // 平均访问次数
      (sex, ageGroup, totalUsers.toString, listingUsers.toString, avgListingVisits)
    }

    // 3. 整理结果格式
    val resultData = windowedData.map { case (sex, ageGroup, totalUsers, listingUsers, avgVisits) =>
      (sex, ageGroup, totalUsers, listingUsers, avgVisits)
    }

    // 4. 写入文件（累加存储，按时间排序）
    resultData.foreachRDD(rdd => {
      val list = new ListBuffer[String]()
      // 添加新数据（按时间排序，键值形式）
      val newData = rdd.sortBy(_._1).collect()  // 按时间升序排列
      newData.foreach { case (sex, ageGroup, totalUsers, listingUsers, avgVisits) =>
        list.append(s"""{"sex": "$sex", "ageGroup": "$ageGroup", "totalUsers": "$totalUsers", "listingUsers": "$listingUsers", "avgListingVisits": "$avgVisits"}""")
        val zhangRealTimeDao = new ZhangRealTimeDao
        zhangRealTimeDao.insertTime4(sex, ageGroup, totalUsers, listingUsers, avgVisits)
      }

      // 写入完整JSON数组（累加历史数据，不覆盖）
      val out = new PrintWriter(new FileWriter(outputFile))
      out.println("[\n" + list.mkString(",\n") + "\n]")
      out.flush()
      out.close()
    })
  }

  //五、用户在不同来源不同设备的产品页次数
  def realTime5(data: DStream[AdClickData]):Unit ={
    // 输出文件路径
    val outputFile = "output/zhangTime5.json"

    // 1. 筛选访问过产品页的用户，提取关键信息
    val productUsers = data
      .filter(_.product_page == "1")  // 仅保留访问过产品页的用户
      .map(line => {
        val userId = line.user_id  // 用户ID，用于去重
        val source = line.source  // 来源渠道
        val device = line.device  // 设备类型
        // 解析总浏览页数（异常值处理为0）
        val totalPages = try { line.total_pages_visited.toInt } catch { case _: Exception => 0 }
        // 组合键：(来源渠道, 设备)；值：(用户ID, 总浏览页数)
        ((source, device), (userId, totalPages))
      })
      // 按组合键聚合，关联同一窗口、渠道、设备的用户数据
      .groupByKey()

    // 2. 统计浏览深度区间分布及占比
    val windowedData = productUsers.map { case ((source, device), userPages) =>
      // 按用户ID去重，保留每个用户的总浏览页数（取最大值，避免重复统计）
      val uniqueUsers = userPages
        .groupBy(_._1)  // 按userId分组
        .mapValues(behaviors => behaviors.map(_._2).max)  // 取该用户的最大浏览页数
        .values  // 仅保留浏览页数

      val totalUsers = uniqueUsers.size  // 该分组的总用户数

      // 统计各浏览深度区间的用户数
      val page0_5 = uniqueUsers.count(_ <= 5)
      val page6_10 = uniqueUsers.count(p => p >= 6 && p <= 10)
      val page11Plus = uniqueUsers.count(_ >= 11)

      // 计算各区间占比（避免除零，总用户为0时占比均为0）
      val ratio0_5 = if (totalUsers > 0) (page0_5.toDouble / totalUsers * 100).formatted("%.2f") else "0.00"
      val ratio6_10 = if (totalUsers > 0) (page6_10.toDouble / totalUsers * 100).formatted("%.2f") else "0.00"
      val ratio11Plus = if (totalUsers > 0) (page11Plus.toDouble / totalUsers * 100).formatted("%.2f") else "0.00"

      (source, device, totalUsers.toString, ratio0_5, ratio6_10, ratio11Plus)
    }

    // 3. 整理结果格式
    val resultData = windowedData.map { case (source, device, totalUsers, r0_5, r6_10, r11Plus) =>
      (source, device, totalUsers, r0_5, r6_10, r11Plus)
    }

    // 4. 写入文件（累加存储，按时间排序）
    resultData.foreachRDD(rdd => {
      val list = new ListBuffer[String]()
      // 添加新数据（按时间排序，键值形式）
      val newData = rdd.sortBy(_._1).collect()  // 按时间升序排列
      newData.foreach { case (source, device, totalUsers, r0_5, r6_10, r11Plus) =>
        list.append(s"""{"source": "$source", "device": "$device", "totalUsers": "$totalUsers", "0-5页占比": "$r0_5%", "6-10页占比": "$r6_10%", "11页+占比": "$r11Plus%"}""")
        val zhangRealTimeDao = new ZhangRealTimeDao
        zhangRealTimeDao.insertTime5(source, device, totalUsers, r0_5, r6_10, r11Plus)
      }

      // 写入完整JSON数组（累加历史数据，不覆盖）
      val out = new PrintWriter(new FileWriter(outputFile))
      out.println("[\n" + list.mkString(",\n") + "\n]")
      out.flush()
      out.close()
    })
  }

  //六、不同用户在不同市场的支付率
  def realTime6(data: DStream[AdClickData]):Unit ={
    // 输出文件路径
    val outputFile = "output/zhangTime6.json"

    // 1. 筛选非新用户且访问过支付页的用户，提取关键信息
    val paymentUsers = data
      .filter(line => line.new_user == "0" && line.payment_page == "1")  // 非新用户+访问支付页
      .map(line => {
        val userId = line.user_id  // 用户ID，用于去重
        val market = line.market  // 市场级别
        val os = line.operative_system  // 操作系统
        // 标记是否完成支付确认（1=完成，0=未完成）
        val isPaymentConfirmed = if (line.payment_confirmation_page == "1") 1 else 0
        // 组合键：(窗口键, 市场, 操作系统)；值：(用户ID, 支付确认标记)
        ((market, os), (userId, isPaymentConfirmed))
      })
      // 按组合键聚合，关联同一窗口、市场、操作系统的用户数据
      .groupByKey()

    // 2. 计算失败率
    val windowedData = paymentUsers.map { case ((market, os), userPayments) =>
      // 按用户ID去重，保留每个用户的支付确认状态（取首次状态，避免重复统计）
      val uniqueUsers = userPayments
        .groupBy(_._1)  // 按userId分组
        .mapValues(behaviors => behaviors.head._2)  // 取该用户的首次支付确认状态
        .values  // 仅保留支付确认标记

      val totalPaymentUsers = uniqueUsers.size  // 访问支付页的总用户数（分母）
      val failedUsers = uniqueUsers.count(_ == 0)  // 未完成支付确认的用户数（分子）

      // 计算失败率（避免除零，总用户为0时失败率为0）
      val failureRate = if (totalPaymentUsers > 0) (failedUsers.toDouble / totalPaymentUsers * 100).formatted("%.2f") else "0.00"

      (market, os, totalPaymentUsers.toString, failedUsers.toString, failureRate)
    }

    // 3. 整理结果格式
    val resultData = windowedData.map { case (market, os, totalUsers, failedUsers, failureRate) =>
      (market, os, totalUsers,failureRate)
    }

    // 4. 写入文件（累加存储，按时间排序）
    resultData.foreachRDD(rdd => {
      val list = new ListBuffer[String]()
      // 添加新数据（按时间排序，键值形式）
      val newData = rdd.sortBy(_._1).collect()  // 按时间升序排列
      newData.foreach { case (market, os, totalUsers,failureRate) =>
        list.append(s"""{"market": "$market", "os": "$os", "totalPaymentUsers": "$totalUsers", "failureRate": "$failureRate%"}""")
        val zhangRealTimeDao = new ZhangRealTimeDao
        zhangRealTimeDao.insertTime6(market, os, totalUsers,failureRate)
      }

      // 写入完整JSON数组（累加历史数据，不覆盖）
      val out = new PrintWriter(new FileWriter(outputFile))
      out.println("[\n" + list.mkString(",\n") + "\n]")
      out.flush()
      out.close()
    })
  }


}
